package edu.gatech.ece6102;

import edu.gatech.ece6102.beans.AddressBean;
import edu.gatech.ece6102.beans.Message;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.LinkedList;
import java.util.Queue;

/**
 * To represent an Application server, we will store all the connection for that particular app server
 * User: deepal
 * Date: Oct 4, 2010
 * Time: 3:25:57 PM
 */
public class AppServer extends Thread {

    //Incomming message queue
    private final Queue<Message> messageQueue;
    // Number of errors
    private int errorCount = 0;
    //Whether reporting error or not (processing failover)
    private boolean reportingError = false;
    private boolean failOver = false;
    //Reference to the logbased executor
    private LogbasedExecutor logbasedExecutor;
    //Addressing information of the app server
    private AddressBean addressBean;
    //An optional name
    private String name;

    public AppServer(AddressBean addressBean,
                     LogbasedExecutor logbasedExecutor, String name) {
        messageQueue = new LinkedList<Message>();
        this.logbasedExecutor = logbasedExecutor;
        this.addressBean = addressBean;
        this.name = name;
        start();
    }

    /**
     * To string method, only for testing purposes
     * @return
     */
    public String toString(){
        return addressBean.getIp() + " " + addressBean.getPort() +  " " + name;
    }

    public boolean isReportingError() {
        return reportingError;
    }

    /**
     * Here we keep looking at the message queue, if it is not empty then poll the message and create a worker and invoke it
     * If the queue is empty then it wait until it gets a notification. (Java wait-and-notify)
     */
    public void run() {
        while (!reportingError) {
            if (messageQueue.isEmpty()) {   // Queue is empty
                try {
                    synchronized (this) {
                        this.wait(); // we wait
                    }

                } catch (InterruptedException e) {
                    continue;
                }
            }
            // Has a message
            Message message = messageQueue.poll();
            //Create a worker
            Thread worker = new Thread(new Worker(message));
            worker.start();
        }
    }

    /**
     * Adding a message to message queue by the logbasedExcutor
     * @param message was able to add or not
     * @return
     */
    public boolean addMessage(Message message) {
        //App server is processing error, so cannot take any more messages
        if (reportingError) {
            System.out.println("Reporting an error");
            return false;
        }
        //Adding the message to the queue
        messageQueue.add(message);
        synchronized (this) {
            //notify the threads waiting for the queue
            this.notify();
        }
        return true;
    }

    /**
     * update the logbasedexecutor about the error and add the message back to the queue
     */
    private void reportError() {
        if (!failOver) {
            failOver = true;
            logbasedExecutor.removeAppServer(this);
            System.out.println("Done fail over: " + name);
            return;
        }

    }


    /**
     * Message invoking worker
     */
    class Worker implements Runnable {
        Message message;
        Worker(Message message) {
            this.message = message;

        }

        public void run() {
            try {
                //Create a socket to connect to app server
                Socket socketToServer = new Socket(InetAddress.getByName(addressBean.getIp()), addressBean.getPort());
                OutputStream out = socketToServer.getOutputStream();
                InputStream in = socketToServer.getInputStream();
                if (message == null) {
                    System.out.println("MESSAGE NULL");
                }
                //send the message to app server
                out.write(message.getMessage());
                out.flush();
                //write response to the client
                boolean status = writeToClient(in, message.getSocket().getOutputStream(), socketToServer);
                if (status) {
                    message.getSocket().getInputStream().close();
                    //remove the message from the main queue
                    logbasedExecutor.removeFromQueue(message);
                    //remove the message from the logger
                    logbasedExecutor.removeFromLog(message);
                } else {
                    //if there is an error
                    errorCount++;
                    // need to handle error
                    handleError();
                }

            } catch (IOException e) {
                System.out.println("FOUND ERROR");
                //Exception, so incrementing the counter
                errorCount++;
                handleError();
            }
        }

        /**
         * If we have encountered more than two errors then we report error
         */
        private void handleError() {
            System.out.println("Handle Error " + name);
            logbasedExecutor.addToQueue(message);
            if (errorCount > 2) {
                reportingError = true;
                reportError();
            }
        }


        /**
         * Writing the response to the client
         * @param in  InputStream (from App Server)
         * @param out Output stream (Client)
         * @param socket  Client socket
         * @return was able to process it or not
         */
        private boolean writeToClient(InputStream in, OutputStream out, Socket socket) {
            byte[] buf = new byte[2048];
            int n;
            try {
                while ((n = in.read(buf)) > 0) {
                    out.write(buf, 0, n);
                    out.flush();
                    if (!socket.isConnected()) {
                        break;
                    }
                }
            }catch (SocketException se){
               if(se.getMessage().startsWith("Connection reset by peer")){
                   System.out.println("Found error on Tomcat");
               }
            } catch (IOException e) {
                //could not able to write the response
                e.printStackTrace();
                return false;

            } finally {
                try {
                    in.close();
                } catch (IOException e) {
                }
            }
            return true;
        }
    }
}
